package org.example.reactive;

import java.util.concurrent.Flow;
import java.util.concurrent.SubmissionPublisher;

/**
 * @author zhaorui
 * @Date 2024/6/15  9:26
 * @description:
 */
public class ReactiveProcessor extends SubmissionPublisher<String> implements Flow.Processor<String, String> {
    // 保存订阅关系
    Flow.Subscription sub;

    @Override
    public void onSubscribe(final Flow.Subscription subscription) {
        this.sub = subscription;

        // 第一次需要
        sub.request(1);
        System.out.println("*** 发送者 *** 建立订阅关系");
    }

    @Override
    public void onNext(final String item) {
        System.out.println("*** 发送者 *** 接收到的数据: " + item);

        // 一些业务处理
        this.submit(item+"*** 发送者 ***");//比如发送数据给接收者

        // 消费者需要接收多少数据，是消费者通过这种方式告诉发送者，这样就可以避免消费者被压垮
        sub.request(10);
    }

    @Override
    public void onError(final Throwable throwable) {
        System.out.println("*** 发送者 *** 发生了错误:" + throwable.getMessage());
    }

    @Override
    public void onComplete() {
        System.out.println("*** 发送者 *** 数据发送完会触发的方法");
    }
}
